package com.chb.sparkstreaming

import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.kafka010._
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe

object CreateDirectDStream {
    def main(args: Array[String]): Unit = {

        val conf = new SparkConf().setAppName("test").setMaster("local[*]")
        val streamingContext = new StreamingContext(conf, Seconds(1))
        val kafkaParams = Map[String, Object](
            "bootstrap.servers" -> "10.0.0.201:9092",
            "key.deserializer" -> classOf[StringDeserializer],
            "value.deserializer" -> classOf[StringDeserializer],
            "group.id" -> "use_a_separate_group_id_for_each_stream",
            "auto.offset.reset" -> "latest",
            "enable.auto.commit" -> (false: java.lang.Boolean)
        )

        val topics = Array("test")
        val stream = KafkaUtils.createDirectStream[String, String](
            streamingContext,
            PreferConsistent,  // 任务尽量均匀分布在各个executor节点
            Subscribe[String, String](topics, kafkaParams)
        )

        stream.map(record => (record.key, record.value))
          .foreachRDD(_.foreach(println))
        streamingContext.start()
        streamingContext.awaitTermination()  // 如果没有这个，会直接中止
    }

}
